package fenxiTuijian

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, second}

object etl {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("test")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    // todo 首先读取数据然后下面对数据进行预处理
    var order = spark.table("work.order_index")
    var median=spark.table("work.media_index")

    //  todo 1） 所有表做数据去重处理；
    println(" 1） 所有表做数据去重处理；")
    println("order表去除重复数据前的数量",order.count())
    println("median表去除重复数据前的数量",median.count())
    order=order.dropDuplicates()
    median=median.dropDuplicates()

    println("order表去除重复数据后的数量", order.count())
    println("median表去除重复数据后的数量", median.count())

    //  todo （2） 删除所有表中owner_name等于“EA级”、“EB级”、“EC级”、“ED级”或“EE级”的数据；
    println("（2）删除所有表中owner_name等于“EA级”、“EB级”、“EC级”、“ED级”或“EE级”的数据；")
    order=order.filter(col("owner_name")!=="EA级")
      .filter(col("owner_name")!=="EB级")
      .filter(col("owner_name")!=="EC级")
      .filter(col("owner_name")!=="ED级")
      .filter(col("owner_name")!=="EE级")

    median=median.filter(median("owner_name")!=="EA级")
      .filter(col("owner_name")!=="EB级")
      .filter(col("owner_name") !== "EC级")
      .filter(col("owner_name") !== "ED级")
      .filter(col("owner_name") !== "EE级")

    order.select("owner_name").show()
    median.select("owner_name","station_name").show()

    //  todo （3） 删除所有表中owner_code等于02、09或10的数据；
    println("（3）删除所有表中owner_code等于02、09或10的数据；")
    order=order.filter(col("owner_code")!=="02")
      .filter(col("owner_code")!=="09")
      .filter(col("owner_code")!=="10")

   median = median.filter(col("owner_code") !== "02")
      .filter(col("owner_code") !== "09")
      .filter(col("owner_code") !== "10")

    order.select("owner_code").show()
    median.select("owner_code","station_name").show()

    //  todo （4） 删除用户收视行为信息表中观看时长小于20秒（duration<20000）的数据；
    println("（4） 删除用户收视行为信息表中观看时长小于20秒（duration<20000）的数据；")
    median=median.filter(col("duration") > 20000)

    median.select("duration","station_name").show()

    //  todo （5） 删除用户收视行为信息表中观看时长大于5小时（duration>18000000）的数据；
    println("（5） 删除用户收视行为信息表中观看时长大于5小时（duration>18000000）的数据；")
    median=median.filter(col("duration") < 18000000)

    median.select("duration","station_name").show()

    //  todo 6） 只保留所有表中业务品牌（sm_name）为珠江宽频、互动电视、甜果电视、数字电视的数据；
    println("6） 只保留所有表中业务品牌（sm_name）为珠江宽频、互动电视、甜果电视、数字电视的数据；")
    val sm_name_list = Seq("珠江宽频", "互动电视", "甜果电视", "数字电视")
    order=order.filter(col("sm_name").isin(sm_name_list: _*))

    median = median.filter(col("sm_name").isin(sm_name_list: _*))

    order.select("sm_name").show()
    median.select("sm_name","station_name").show()

    //  todo 7） 只保留所有表中状态名称（run_name)为正常、主动暂停、欠费暂停、主动销户的数据；
    println("7）只保留所有表中状态名称（run_name)为正常、主动暂停、欠费暂停、主动销户的数据；")
    order=order.filter(
      col("run_name")==="正常" or col("run_name")==="主动暂停"
      or col("run_name")==="欠费暂停" or col("run_name")==="主动销户"
    )
//      .filter(col("run_name")==="主动暂停")
//      .filter(col("run_name")==="欠费暂停")
//      .filter(col("run_name")==="主动销户")

    order.select("run_name").show()

    //  todo （8） 删除用户收视行为信息表中res_type=0且origin_time和end_time中秒单位为00的数据。
    println("（8） 删除用户收视行为信息表中res_type=0且origin_time和end_time中秒单位为00的数据。")
    median=median.filter(
      (col("res_type")===0).and(second(col("origin_time"))!==0)
      .and(second(col("end_time"))!==0)
    )


//    median=median.filter(col("res_type")=!=0)

    median.select("res_type","origin_time","end_time","station_name").show()


    //  todo 将处理好的数据写入hive
    order.write.mode("overwrite").saveAsTable("work.order_clean")
    median.write.mode("overwrite").saveAsTable("work.media_clean")











    spark.close()
  }

}
